package com.udf.flink.udf;

import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class testA extends ScalarFunction {
    private static String driverName = "oracle.jdbc.driver.OracleDriver";
    private static String dbUrl = "jdbc:oracle:thin:@118.31.39.174:1521:xe";
    private static String username = "study";
    private static String password = "study";
    private transient Connection connection;
    private transient PreparedStatement preparedStatement;
    private transient PreparedStatement preparedStatementHeartbeat;
    private String sql = "insert into AAAA values(?,?)";
    private String sqlHeartbeat = "select 1 from dual";

    private static final Logger LOG = LoggerFactory.getLogger(testA.class);

    @Override
    public void open(FunctionContext context){
        try{
            this.connection = establishConnection();
            preparedStatement = connection.prepareStatement(sql);
            preparedStatementHeartbeat = connection.prepareStatement(sqlHeartbeat);

            new ScheduledThreadPoolExecutor(1).scheduleAtFixedRate(
                    () -> HeartbeatTest(),
                    1800000,
                    1800000,
                    TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            LOG.error(e.getMessage(),e);
        }
    }

    //查不到该账户和证件，需新增uc_no并插入表中
    public boolean eval(String id, String name){
        checkConnectionOpen(this.connection);
        int rs = 0;
        /*
        try {
            Thread.sleep(30000); //测试结果：每次只有一条数据调用eval
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
         */
        try {
            preparedStatement.clearParameters();
            preparedStatement.setString(1,id);
            preparedStatement.setString(2,name);
            rs = preparedStatement.executeUpdate();
            connection.commit();
        }catch(Exception e){
            LOG.error(e.getMessage(),e);
        }finally {
            return (rs > 0 ? true:false);//是否执行成功
        }
    }

    private void HeartbeatTest() {
        try{
            preparedStatementHeartbeat.executeQuery();
        } catch (SQLException e) {
            LOG.error(e.getMessage(),e);
            checkConnectionOpen(connection);
        }
    }

    private void checkConnectionOpen (Connection conn) {
        try{
            if(conn.isClosed()){
                conn = establishConnection();
                this.connection = conn;
                preparedStatement = connection.prepareStatement(sql);
                preparedStatementHeartbeat = connection.prepareStatement(sqlHeartbeat);
            }
        } catch (SQLException | ClassNotFoundException e){
            LOG.error(e.getMessage(),e);
        }
    }

    private Connection establishConnection() throws ClassNotFoundException, SQLException {
        Class.forName(driverName);
        connection = DriverManager.getConnection(dbUrl, username, password);
        return connection;
    }

    @Override
    public void close(){
        try {
            if (preparedStatement != null || (!preparedStatement.isClosed()) ) {
                preparedStatement.close();
                preparedStatement = null;
            }
            if (preparedStatementHeartbeat != null || (!preparedStatementHeartbeat.isClosed()) ) {
                preparedStatementHeartbeat.close();
                preparedStatementHeartbeat = null;
            }
            if(connection != null || (!connection.isClosed()) ){
                connection.close();
                connection = null;
            }
        } catch (Exception e) {
            LOG.error(e.getMessage(),e);
        }
    }
}
